package com.health.trans.kit;

import cn.hutool.core.date.DateUtil;
import cn.hutool.core.util.ObjectUtil;
import com.health.trans.job.AbstractJob;
import com.health.trans.pojo.*;
import com.health.trans.util.JobUtil;
import lombok.extern.slf4j.Slf4j;

import java.io.Serializable;
import java.util.ArrayList;
import java.util.Date;
import java.util.List;
import java.util.Objects;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.LongAdder;
import java.util.function.Consumer;
import java.util.stream.Collectors;

/**
 * @author nqlz
 */
@Slf4j
public class BatchSendInfo<T, J extends Serializable> extends SendBase<T, J> {
    private final TableRes res = new TableRes();
    /**
     * 子任务总完成数量
     */
    private final LongAdder index = new LongAdder();
    /**
     * 所有任务开始时间
     */
    private final Date solveStartDate = new Date();

    protected List<JobConfig<T>> jobConfigs = new ArrayList<>();

    public BatchSendInfo(TransParam param, TableInfo<T> tableInfo) {
        super(param, tableInfo);
        initJobConfig();
    }

    public void initJobConfig() {
        Consumer<JobConfig<T>> consumer = AbstractJob.getSolveConsume(param, tableInfo);
        String otherCondition = baseSql + conditionBuilder.toString();
        //当无时间优化时
        if (ctTimeDtos.stream().filter(Objects::nonNull).count() < 1 || targetCount <= optimizeErBai) {
            getJcOneTask(consumer, otherCondition);
            return;
        }
        //分解任务
        AtomicInteger task = new AtomicInteger(1);
        jobConfigs = ctTimeDtos.stream()
                .map(i -> {
                    JobConfig<T> tJobConfig = getJobConfig(task.get(), i.getCount(), baseSql + getSqlByCts(i));
                    task.getAndIncrement();
                    return tJobConfig;
                })
                .peek(consumer)
                .collect(Collectors.toList());
    }

    private void getJcOneTask(Consumer<JobConfig<T>> consumer, String otherCondition) {
        JobConfig<T> tJobConfig = getJobConfig(1, queryCount, otherCondition);
        consumer.accept(tJobConfig);
        jobConfigs.add(tJobConfig);
    }

    private JobConfig<T> getJobConfig(int taskIndex, long taskCount, String sql) {
        JobConfig<T> instance = JobConfig.getInstance(tableInfo, param);
        ReaderConfig<T> readerConfig =param.getReader().getReaderConfig();
        readerConfig.setSql(baseSql);
        readerConfig.setTableName(tableName);
        readerConfig.setTableClass(tableInfo.getReadTableClass());
        tableInfo.setReaderConfig(readerConfig);

        WriterConfig<T> writerConfig = param.getWriter().getWriterConfig();
        writerConfig.setJobConfig(instance);
        tableInfo.setWriterConfig(writerConfig);

        return instance.setTaskCount(taskCount)
                .setSql(sql)
                .setTaskIndex(taskIndex)
                .setTableName(getTableName())
                .setQueryCount(queryCount);
    }


    public String getSqlByCts(CtTimeDto ctTimeDto) {
        String sql = " (" + "" + optimizedUnderName + "" + " >= '"
                + DateUtil.formatDateTime(ctTimeDto.getMinDate()) + "' and  " + optimizedUnderName + " < '"
                + DateUtil.formatDateTime(ctTimeDto.getTargetDate()) + "')";
        return BatchConditionKit.appendSql(ObjectUtil.clone(conditionBuilder), sql);
    }

    public TableRes sendAction() {
        log.info("开始处理：" + tableName);
        jobConfigs.forEach(jobConfig -> {
            jobConfig.setChunkSize(param.getReader().getChunkSize());
            jobConfig.setMaxCount((long) param.getReader().getMaxCount());
            jobConfig.setAllDoneCount(index);
            jobConfig.setTotalTask(jobConfigs.size());
            JobUtil.execute(jobConfig.getJobInit().getJob());
        });

        String solveTime = ThreadMonitor.getTime(solveStartDate, new Date());
        String allTime = ThreadMonitor.getTime(initStartDate, new Date());
        log.info("监控到任务已完成。总时间时间；{},处理时间为：{}", allTime, solveTime);
        res.setCount(param.getReader().getSource() + "->"  + " 已处理" + index.longValue() + "/ 要求" + targetCount + " 总数/" + this.queryCount)
                .setWaitTime(allTime).setTableName(tableInfo.getReadTableClass().getSimpleName());

        return res;
    }

}
